package com.donglink.test.storm.wordcount;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.*;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.windowing.TupleWindow;

import java.util.Map;

public class SplitBolt extends BaseRichBolt {


    ///自我理解：outputCollector是框架传过来的一个collector
    /*


    程序中使用一个变量接受，在不断循环的execute方法中传输Values（相当于一个集合），每次调用emit就会传输一个Values，那么Values就会对应
    声明字段，一一对应传输出去
    Values（“dong”,1,2）
    对应
    outputFieldsDeclarer.declare(new Fields("word","count","num"));

    在下一个bolt中处理时执行一次execute，里面的tuple中就对应这个上个bolt传输过来的一个Values
    通过get方法获取其中传输的属性值
     */

    OutputCollector outputCollector;
    ///初始化方法
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

        this.outputCollector=outputCollector;

    }

    //执行方法
    @Override
    public void execute(Tuple tuple) {
        String line=tuple.getStringByField("line");

        String[] words=line.split(" ");
        for (String word : words)
        {
            outputCollector.emit(new Values(word,1));
        }
    }

    //声明属性
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word","count"));
    }
}
